package com.cd.system.service.impl;

import com.cd.common.domain.GoodOrders;
import com.cd.system.jpa.entity.JavaStudyRecord;
import com.cd.system.jpa.repository.GoodOrderRepository;
import com.cd.system.jpa.repository.JavaStudyRecordRepo;
import com.cd.system.jpa.repository.ProductRepository;
import com.cd.system.jpa.repository.SysMenuRepository;
//import com.cd.system.jpa.repository.TestRepository;
//import com.cd.system.mapper.TestMapper;
import com.cd.common.domain.ResponseResult;
import com.cd.common.exception.LqqException;
import com.cd.system.service.TestService;
import com.cd.common.util.AuthUtils;
import com.cd.common.util.LqqGoodEnum;
import com.cd.common.util.RedisCache;
import com.cd.common.util.StatsReportExcelUtils;
//import com.cd.common.util.kafka.KafkaProducerUtils;
//import org.apache.kafka.clients.admin.AdminClient;
//import org.apache.kafka.clients.admin.AdminClientConfig;
//import org.apache.kafka.clients.admin.ListTopicsOptions;
//import org.apache.kafka.clients.admin.ListTopicsResult;
//import org.apache.kafka.clients.admin.NewTopic;
//import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import javax.servlet.http.HttpServletResponse;
import java.util.Collections;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;

@Service
//@Scope("prototype") //可以把该实例变成多例
public class TestServiceImpl implements TestService {

    private static final Logger logger = LoggerFactory.getLogger(TestServiceImpl.class);

    private static final String LQQ_SEMAPHORE_TEST_STRING="lqqSemaphoreTest";
//    @Autowired
//    TestRepository testDao;
//    @Autowired
//    TestMapper testMapper;
    @Autowired
    ProductRepository productRepository;
    @Autowired
    GoodOrderRepository ordersRepository;
    @Autowired
    SysMenuRepository sysMenuRepository;
    @Autowired
    JavaStudyRecordRepo javaStudyRecordRepo;

    @Autowired
    RedisCache redisCache;

    @Value("${lqq.kafka.ipHost}")
    private String kafkaIpHost;

    private Semaphore semaphore=new Semaphore(5);
    private AtomicInteger counter =new AtomicInteger(2);

//    @Override
//    public String getTestId() {
//        return testDao.findById(1L).toString();
//    }
//
//    @Override
//    public String getTestIdUseMybatis() {
//        return testMapper.selectPermsByUserId(1L);
//    }

    @Override
    @Async
    public String testSync() {

        try {
            Thread.sleep(6000);
            logger.info("sync");
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "sync";
    }

    @Override
    @Async
    public Future<String> testSync1() {
        logger.info("sync1");
        try {
            Thread.sleep(6000);
            logger.info("sync1");
        } catch (InterruptedException e) {
            logger.error("sync1：", e);
            return new AsyncResult<String>("任务一失败");
        }
        return new AsyncResult<String>("任务一完成");
    }

    @Override
    @Async
    public Future<String> testSync2() {
        logger.info("sync2");
        try {
            Thread.sleep(6000);
            logger.info("sync2");
        } catch (InterruptedException e) {
            logger.error("sync2：", e);
            return new AsyncResult<String>("任务2失败");
        }
        return new AsyncResult<String>("任务2完成");
    }

    @Override
    @Async
    public Future<String> testSync3() {
        logger.info("sync3");
        try {
            Thread.sleep(6000);
            logger.info("sync3");
        } catch (InterruptedException e) {
            logger.error("sync3：", e);
            return new AsyncResult<String>("任务3失败");
        }
        return new AsyncResult<String>("任务3完成");
    }

    @Override
    public ResponseResult seckill(Long goodId) {
        //需要手动设置库存 set product:stock:1 2688
        //查询商品数量 小于1则返回失败
//        Product product = productRepository.findByProductId(goodId);
//        if (product.getStock() < 1) {
//            return new ResponseResult(LqqGoodEnum.LQQ_GOOD_NOT_ENOUGH);
//        }
        //商品数量扣减 不足则返回失败
        //-- 传统数据库写法 下面为redis写法
//        try {
//            int i = productRepository.updateByProductId(goodId);
//            if (i < 1) {
//                return new ResponseResult(LqqGoodEnum.LQQ_GOOD_NOT_ENOUGH);
//            }
//            //订单生成 使用mq 这种情况会有超卖的情况
//            GoodOrders orders=new GoodOrders();
//            orders.setGoodsId(goodId);
//            orders.setUserId(1L);
//            orders.setOrderStatus(1);
//            orders.setOrderCreateTime(new Date());
//            orders.setOrderUpdateTime(new Date());
//            ordersRepository.save(orders);
//        }catch (Exception e) {
//            logger.error("扣减库存失败：", e);
//            return new ResponseResult(LqqGoodEnum.LQQ_GOOD_NOT_ENOUGH);
//        }

        //redis扣减库存写法
            String productStockKey = "product:stock:" + goodId;
            Object productStock = redisCache.getCacheObject(productStockKey);

            if (Integer.parseInt(productStock.toString()) < 1) {
                throw new LqqException(LqqGoodEnum.LQQ_GOOD_NOT_ENOUGH.getMsg());
            }
            long stockCount = redisCache.decrement(productStockKey, 1);
            if (stockCount < 0) {
                redisCache.increment(productStockKey, 1);
                //return new ResponseResult(LqqGoodEnum.LQQ_GOOD_NOT_ENOUGH);
                throw new LqqException(LqqGoodEnum.LQQ_GOOD_NOT_PERMISSION.getMsg());
            }
//            UPDATE lqqdb_product SET stock=3000 WHERE id = 1;
//            SELECT COUNT(*) FROM lqqdb_orders;
//            TRUNCATE TABLE lqqdb_orders;
            // 订单生成 使用mq 这种情况会有超卖的情况
        try {
            GoodOrders orders=new GoodOrders();
            orders.setGoodsId(goodId);
            orders.setUserId(AuthUtils.currentUserId());
            orders.setOrderStatus(1);
            orders.setOrderCreateTime(new Date());
            orders.setOrderUpdateTime(new Date());

            //kafka写法
            //如果kafka没有存活 记得报错
            //kafka创建主题
            //createKafkaTopic("goodKill");
            //KafkaProducerUtils.getGoodOrdersKafkaProducer().send(new ProducerRecord("goodKill", orders.toString()));
            //数据库写法
            ordersRepository.save(orders);
        } catch (Exception e) {
            logger.error("kafka：{}", e.getLocalizedMessage());
            //return new ResponseResult(LqqGoodEnum.LQQ_GOOD_NOT_ENOUGH);
            redisCache.increment(productStockKey, 1);
            throw new LqqException("内部异常");

        }
        return new ResponseResult(LqqGoodEnum.LQQ_GOOD_SUCCESS);
    }

    private void createKafkaTopic(String topicName) throws ExecutionException, InterruptedException {
//        // 配置连接到Kafka集群的参数
//        Properties config = new Properties();
//        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,kafkaIpHost);
//
//        // 创建AdminClient实例
//        AdminClient admin = AdminClient.create(config);
//
//        //如果有主题就直接返回 没有主题就创建一个
//        ListTopicsOptions options = new ListTopicsOptions();
//        options.listInternal(false); // 不包含内部主题
//        ListTopicsResult topics = admin.listTopics(options);
//        Set<String> topicNames = topics.names().get();
//        if (topicNames.contains(topicName))
//            return;
//
//        // 设置主题名称及分区数、副本因子
//        int numPartitions = 1;
//        short replicationFactor = 1;
//
//        // 创建NewTopic对象
//        NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
//
//        // 提交创建主题的请求
//        admin.createTopics(Collections.singleton(newTopic)).all().get();
//
//        // 关闭AdminClient实例
//        admin.close();
    }

    @Override
    public void excelImport(HttpServletResponse response) throws IllegalAccessException {
        //请设置表头的行数 如一行两行或三行
//        int rowNum = 1;
//        Map<String, Object> titles = new LinkedHashMap<>();
//        Map<String, String> subTitles = new LinkedHashMap<>();
//        titles.put("departmentName", "所属部门");
//        titles.put("systemName", "系统名称");
//        titles.put("systemAlias", "系统简称");
//        subTitles.put("ActiveUser", "有效用户数");
//        subTitles.put("NewTaskNum", "新建任务数");
//        subTitles.put("RejectedTaskNum", "退回任务数");
//        subTitles.put("CompletedTaskNum", "完成任务数");
//        titles.put("month6", subTitles);
//        titles.put("year8", subTitles);
//        List<String> ss=new LinkedList<>();
//        ss.add("lqq");
//        ss.add("lqq1");
//        ss.add("lqq2");
        //这是什么？
        //StatsReportExcelUtils.menuList(response,menus);
        //http://127.0.0.1:10280/lqq/test/excel/import

        //表头行数
        int rowNum = 1;
        Map<String, Object> titles = new LinkedHashMap<>();
        titles.put("id", "id");
        titles.put("name", "名称");
        titles.put("completion", "完成度");
        titles.put("remark", "备注");

        List<JavaStudyRecord> javaStudyRecordList = javaStudyRecordRepo.findAll();
        StatsReportExcelUtils.export(response, titles, StatsReportExcelUtils.listObjectToListMap(javaStudyRecordList), "测试excel");

    }

    //互斥锁 会让请求大量堆积 导致服务器无法接受相应 不能用于此处
    @Override
    public String testSemaphore() {
//        try {
//            semaphore.acquire();
//
//            try {
//                if (redisCache.getCacheObject(LQQ_SEMAPHORE_TEST_STRING)==null || redisCache.getCacheObject(LQQ_SEMAPHORE_TEST_STRING).toString().length()<1){
//                    logger.error("无缓存，写入中");
//                    Thread.sleep(60000);
//                    redisCache.setCacheObject(LQQ_SEMAPHORE_TEST_STRING,"I AM LQQ");
//                    logger.info("写入成功");
//                }else {
//                    logger.info("cache starting");
//                    return redisCache.getCacheObject(LQQ_SEMAPHORE_TEST_STRING).toString();
//                }
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e.getLocalizedMessage());
//            }finally {
//                semaphore.release();
//            }
//
//            //redisCache.setCacheObject("lqqSemaphoreTest","i am lqq");
//
//
//        }catch (InterruptedException e){
//            logger.error(e.getLocalizedMessage());
//            logger.error("信号量满");
//            return e.getLocalizedMessage();
//        }
//        return "不可能到达的地方";
        //上面的为旧版信号量 已证实不合适 下面仅验证原子能否实现
        if (counter.get()>0) {
            try {
                if (redisCache.getCacheObject(LQQ_SEMAPHORE_TEST_STRING) == null || redisCache.getCacheObject(LQQ_SEMAPHORE_TEST_STRING).toString().length() < 1) {
                    logger.info("当前计数为"+counter.getAndDecrement());
                    logger.error("无缓存，写入中");
                    Thread.sleep(60000);
                    redisCache.setCacheObject(LQQ_SEMAPHORE_TEST_STRING, "I AM LQQ");
                    logger.info("写入成功");
                    counter.incrementAndGet();

                } else {
                    logger.info("cache starting");
                    return redisCache.getCacheObject(LQQ_SEMAPHORE_TEST_STRING).toString();
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e.getLocalizedMessage());
            }
        }else {
            logger.info("当前计数为（满了之后）"+counter.get());
            logger.error("程序已满");
            //加返回提示
            throw new RuntimeException("数据维护中，请稍后使用");
        }

        return "不可能到达的地方";
        }
}
